package com.idanchuang.component.mq.amqp.rabbit;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.HashMap;
import java.util.concurrent.CountDownLatch;

/**
 * @author yjy
 * Created at 2022/3/23 1:56 下午
 */
public class ComplexConsumer {

    private static final CountDownLatch LATCH = new CountDownLatch(1);

    public static void main(String[] args) throws Exception {
        // 通过ConnectionFactory创建Connection
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setUsername("dc_base");
        connectionFactory.setPassword("JFgs33#s3");
        connectionFactory.setHost("10.80.20.209");
        connectionFactory.setVirtualHost("/");
        Connection connection = connectionFactory.newConnection();

        // 声明交换机
        String exchange = "python-test-exchange";
        Channel channel = connection.createChannel();
        channel.exchangeDeclare(exchange, "direct");

        // 绑定键
        String bindingKey = "tester";

        // 通过channel声明一个队列，并绑定至交换机
        channel = connection.createChannel();
        String queue1 = "amqp-test-queue1-" + System.currentTimeMillis();
        channel.queueDeclare(queue1, false, true, true, new HashMap<>());
        channel.queueBind(queue1, exchange, bindingKey);
        // 通过这个channel中订阅声明的queue，并通过多线程添加两个消费者，实现并发消费
        startConsume(queue1, new MyConsumer(channel), channel);
        startConsume(queue1, new MyConsumer(channel), channel);

        // 通过一个新的channel声明一个新的队列，并绑定至交换机
        channel = connection.createChannel();
        String queue2 = "amqp-test-queue2-" + System.currentTimeMillis();
        channel.queueDeclare(queue2, false, true, true, new HashMap<>());
        channel.queueBind(queue2, exchange, bindingKey);
        // 通过这个channel中订阅声明的queue，并配置一个消费者
        startConsume(queue2, new MyConsumer(channel), channel);

        System.out.println("Listening...");
        LATCH.await();
    }

    private static void startConsume(String queue, Consumer consumer, Channel channel) {
        Thread thread1 = new Thread(() -> {
            try {
                channel.basicConsume(queue, false, consumer);
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
        thread1.start();
    }

    private static class MyConsumer extends DefaultConsumer {

        public MyConsumer(Channel channel) {
            super(channel);
        }

        @Override
        public void handleDelivery(String consumerTag,
                                   Envelope envelope,
                                   AMQP.BasicProperties properties,
                                   byte[] body) throws IOException
        {
            // 处理消息
            System.out.println(getConsumerTag() + " > 收到消息: " + new String(body));
            long deliveryTag = envelope.getDeliveryTag();
            getChannel().basicAck(deliveryTag, false);
        }

    }
}
